package cn.redragon.soa.common.service;


import static cn.redragon.soa.common.constant.CacheConstants.CMD_DEL;
import static cn.redragon.soa.common.constant.CacheConstants.CMD_HGETALL;
import static cn.redragon.soa.common.constant.CacheConstants.CMD_KEYS;

import cn.redragon.soa.common.dto.GeoDTO;
import java.math.BigDecimal;
import java.time.Duration;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import javax.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.data.geo.Circle;
import org.springframework.data.geo.Distance;
import org.springframework.data.geo.Metrics;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.connection.lettuce.LettuceConverters;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.GeoOperations;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component;

@Component
@RequiredArgsConstructor
@ConditionalOnClass(RedisConnection.class)
@Slf4j
public class RedisService {

    private static final byte[] UNLOCK_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end".getBytes();
    private static final byte[] UNLOCK_SCRIPT_EXPIRE = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('pexpire', KEYS[1], 0) else return 0 end".getBytes();
    @Qualifier("stringRedisTemplate")
    private final RedisTemplate<String, String> template;
    private ValueOperations<String, String> valueOps;
    private ListOperations<String, String> listOps;
    private HashOperations<String, String, String> hashOps;
    private SetOperations<String, String> setOps;
    private ZSetOperations<String, String> zsetOps;
    private GeoOperations<String, String> geoOperations;
    @Value("${spring.cache.hGetAll-size-limit:5000}")
    private Long hGetAllSizeLimit;
    @Value("#{'${spring.cache.disabled-cmd:flushall,flushdb,keys}'.split(',')}")
    private List<String> disabledCmd;

    /**
     * @since 2023-03
     */
    @Deprecated
    public Set<String> keys(String pattern) {
        if (disabledCmd.contains(CMD_KEYS)) {
            throw new UnsupportedOperationException();
        }
        return this.template.keys(pattern);
    }

    public String get(String key) {
        return this.valueOps.get(key);
    }

    public List<String> multiGet(Collection<String> keys) {
        return this.valueOps.multiGet(keys);
    }

    public void set(String key, String value) {
        this.valueOps.set(key, value);
    }

    public void setEx(String key, String value, long timeout) {
        this.valueOps.set(key, value, timeout, TimeUnit.SECONDS);
    }

    public void multiSet(Map<String, String> map) {
        this.valueOps.multiSet(map);
    }

    public Long increase(String key, long delta) {
        return this.valueOps.increment(key, delta);
    }

    public void leftPush(String key, String value) {
        this.listOps.leftPush(key, value);
    }

    public String rightPopAndLeftPush(String key, String value) {
        return listOps.rightPopAndLeftPush(key, value);
    }

    public void trim(String key, long start, long end) {
        listOps.trim(key, start, end);
    }

    public Long size(String key) {
        return listOps.size(key);
    }

    public Long remove(String key, long start, String value) {
        return listOps.remove(key, start, value);
    }

    public void leftPushIfPresent(String key, String value) {
        this.listOps.leftPushIfPresent(key, value);
    }

    public List<String> range(String key, long start, long end) {
        return listOps.range(key, start, end);
    }

    public String index(String key, long index) {
        return listOps.index(key, index);
    }

    public String rightPop(String key) {
        return listOps.rightPop(key);
    }

    public Long leftPushAll(String key, List<String> values) {
        return this.listOps.leftPushAll(key, values);
    }

    public Long rightPush(String key, String value) {
        return this.listOps.rightPush(key, value);
    }

    public void hashSet(String key, String field, Object value) {
        this.hashOps.put(key, field, value.toString());
    }

    public Boolean hashPutIfAbsent(String key, String field, Object value) {
        return this.hashOps.putIfAbsent(key, field, value.toString());
    }

    public Long hashDel(String key, String field) {
        return this.hashOps.delete(key, field);
    }

    public String hashGet(String key, String field) {
        return this.hashOps.get(key, field);
    }

    public Long hashIncrement(String key, String field, long value) {
        return this.hashOps.increment(key, field, value);
    }

    /**
     * @since 2023-03
     * @deprecated replaced by hashScan
     */
    @Deprecated
    public Map<String, String> hashGetAll(String key) {
        if (disabledCmd.contains(CMD_HGETALL)) {
            log.warn("Switch to hashScan automatically since it's forbidden to use hGetAll for key={}", key);
            return this.hashScan(key);
        }
        Long size = hashOps.size(key);
        if (size > hGetAllSizeLimit) {
            log.warn("Switch to hashScan automatically since it's forbidden to use hGetAll on big map with size={} for key={}", size, key);
            return this.hashScan(key);
        }
        try {
            return this.hashOps.entries(key);
        } catch (Exception e) {
            //fall back to hashScan in case disabled-cmd config is not updated accordingly
            return this.hashScan(key);
        }
    }

    public Map<String, String> hashScan(String key) {
        Map<String, String> result = new HashMap<>();
        ScanOptions options = ScanOptions.scanOptions().count(hGetAllSizeLimit).build();
        Cursor<Map.Entry<String, String>> cursor = this.hashOps.scan(key, options);
        Map.Entry<String, String> entry;
        while (cursor.hasNext()) {
            entry = cursor.next();
            result.put(entry.getKey(), entry.getValue());
        }
        return result;
    }

    public void hashPutAll(String key, Map<String, String> map) {
        this.hashOps.putAll(key, map);
    }

    /**
     * @since 2023-03
     * @deprecated replaced by hashScanKeys
     */
    @Deprecated
    public Set<String> hashKeys(String key) {
        Long size = hashLen(key);
        if (size > hGetAllSizeLimit) {
            log.warn("Switch to hashScan automatically since it's forbidden to use hKeys on big map with size={} for key={}", size, key);
            return this.hashScanKeys(key);
        }
        return this.hashOps.keys(key);
    }

    public Set<String> hashScanKeys(String key) {
        Set<String> result = new HashSet<>();
        ScanOptions options = ScanOptions.scanOptions().count(hGetAllSizeLimit).build();
        Cursor<Map.Entry<String, String>> cursor = this.hashOps.scan(key, options);
        while (cursor.hasNext()) {
            result.add(cursor.next().getKey());
        }
        return result;
    }

    public Long hashLen(String key) {
        return this.hashOps.size(key);
    }

    public Boolean hashExists(String key, Object field) {
        return this.hashOps.hasKey(key, field);
    }


    /**
     * @since 2023-03
     * @deprecated replaced by expireNow
     */
    @Deprecated
    public void del(String key) {
        if (disabledCmd.contains(CMD_DEL)) {
            //fall back to expire now
            this.expireNow(key);
        }
        try {
            this.template.delete(key);
        } catch (Exception e) {
            //fall back to expire now in case disabled-cmd config is not updated accordingly
            this.expireNow(key);
        }
    }

    /**
     * @since 2023-03
     */
    @Deprecated
    public void del(Collection<String> keys) {
        if (disabledCmd.contains(CMD_DEL)) {
            throw new UnsupportedOperationException();
        }
        this.template.delete(keys);
    }

    public Boolean exist(String key) {
        return this.template.execute((RedisConnection redisConnection) -> redisConnection.exists(key.getBytes()));
    }

    public Long ttl(String key) {
        return this.template.execute((RedisConnection redisConnection) -> redisConnection.ttl(key.getBytes()));
    }

    public void expire(String key, long timeout, TimeUnit unit) {
        this.template.expire(key, timeout, unit);
    }

    public void expireNow(String key) {
        this.template.expire(key, 0, TimeUnit.MILLISECONDS);
    }

    public void expiredAt(String key, Date date) {
        this.template.expireAt(key, date);
    }

    /**
     * Acquire lock by SetNxEX
     */
    public boolean acquireLock(String key, String value, Duration duration) {
        return Boolean.TRUE.equals(valueOps.setIfAbsent(key, value, duration));
    }

    /**
     * Release lock by redis lua script.
     */
    public void releaseLock(String key, String value) {
        byte[] scripts;
        if (!disabledCmd.contains(CMD_DEL)) {
            scripts = UNLOCK_SCRIPT;
        } else {
            scripts = UNLOCK_SCRIPT_EXPIRE;
        }
        try {
            template.execute((RedisConnection con) -> con.eval(scripts, ReturnType.INTEGER, 1, key.getBytes(), value.getBytes()));
        } catch (Exception e) {
            if (scripts == UNLOCK_SCRIPT) {
                //fall back to expire now in case disabled-cmd config is not updated accordingly
                template.execute((RedisConnection con) -> con.eval(UNLOCK_SCRIPT_EXPIRE, ReturnType.INTEGER, 1, key.getBytes(), value.getBytes()));
            }
        }
    }

    public void addGeo(String key, Set<RedisGeoCommands.GeoLocation<String>> locations) {
        geoOperations.add(key, locations);
    }

    public void removeGeo(String key, String member) {
        geoOperations.remove(key, member);
    }

    /**
     * List the store number by given radius(in kilometer) from specified location.
     *
     * @param key    cache key
     * @param lng    longitude
     * @param lat    latitude
     * @param radius radius in kilometer
     * @return store number
     */
    public List<GeoDTO> geoRadiusAsc(String key, BigDecimal lng, BigDecimal lat, BigDecimal radius) {
        Circle circle = new Circle(new Point(lng.doubleValue(), lat.doubleValue()), new Distance(radius.doubleValue(), Metrics.KILOMETERS));
        RedisGeoCommands.GeoRadiusCommandArgs args = RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs().sortAscending().includeDistance();

        return geoOperations.radius(key, circle, args).getContent().stream()
                .map(r -> GeoDTO.builder().code(r.getContent().getName()).distance(r.getDistance().getValue()).build())
                .collect(Collectors.toList());
    }

    public boolean selectAndSetEx(Integer database, String key, String value, long timeout) {
        StringBuilder script = new StringBuilder();
        script.append("redis.call('select', ARGV[3]); return redis.call('SETEX', KEYS[1], ARGV[2], ARGV[1])");
        return this.template.execute((RedisConnection connection) -> connection
                .eval(script.toString().getBytes(), ReturnType.BOOLEAN, 1, key.getBytes(),
                        value.getBytes(), LettuceConverters.toBytes(timeout), LettuceConverters.toBytes(database)));
    }

    public boolean selectAndDel(Integer database, String key, String value) {
        StringBuilder script = new StringBuilder();
        script.append("redis.call('select', ARGV[2]); if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end");
        return this.template.execute((RedisConnection connection) -> connection
                .eval(script.toString().getBytes(), ReturnType.BOOLEAN, 1, key.getBytes(),
                        value.getBytes(), LettuceConverters.toBytes(database)));
    }

    public boolean setNxPx(String key, String value, long milliseconds) {
        final String script = "return redis.call('setnx', KEYS[1], ARGV[1]) + redis.call('pexpire', KEYS[1], ARGV[2])";
        return Long.parseLong(this.template.execute((RedisConnection redisConnection) -> redisConnection
                .eval(script.getBytes(), ReturnType.INTEGER, 1, key.getBytes(), value.getBytes(),
                        LettuceConverters.toBytes(milliseconds))).toString()) == 2L;
    }

    public boolean increaseWithCondition(String key, int delta, int limit) {
        final String script = "if redis.call('exists',KEYS[1]) == 1 then  " +
            "if tonumber(redis.call('get',KEYS[1])) >= tonumber(ARGV[2]) then " +
            "return 0;" +
            "else " +
            "redis.call('incrby', KEYS[1], ARGV[1]);" +
            "return 1; end; " +
            "else " +
            "redis.call('incrby', KEYS[1], ARGV[1]);" +
            "return 1; end; ";
        String result = this.template.execute(
            (RedisConnection redisConnection) -> redisConnection
                .eval(script.getBytes(), ReturnType.INTEGER, 1, key.getBytes(), LettuceConverters.toBytes(delta), LettuceConverters.toBytes(limit))).toString();
        return Integer.valueOf(result) == 1;
    }

    public boolean hashIncreaseWithCondition(String key, String field, int initial, int delta) {
        log.info("Hash increase with condition:" + key);;
        final String script = "if redis.call('hexists',KEYS[1],KEYS[2]) == 1 then  " +
            "if tonumber(redis.call('hget',KEYS[1],KEYS[2])) + tonumber(ARGV[2]) >= 0 then " +
            "redis.call('hincrby', KEYS[1], KEYS[2] , ARGV[2]); " +
            "return 1; " +
            "else " +
            "return 0; " +
            "end ;" +
            "else " +
            "if  tonumber(ARGV[1]) + tonumber(ARGV[2]) >= 0 then " +
            "redis.call('hincrby', KEYS[1], KEYS[2] , ARGV[1] + ARGV[2]); " +
            "return 1; " +
            "else " +
            "return 0; " +
            "end ;" +
            "end;";
        String result = this.template.execute(
            (RedisConnection redisConnection) -> redisConnection
                .eval(script.getBytes(), ReturnType.INTEGER, 2, key.getBytes(), field.getBytes(), LettuceConverters.toBytes(initial),
                      LettuceConverters.toBytes(delta))).toString();
        return Integer.valueOf(result) == 1;
    }

    public List<String> batchRetrieve(List<String> keys) {
        if (keys.size() > 50) {
            log.warn("Batch retrieve size is {}, beware of redis performance degradation!", keys.size());
        }
        StringBuilder keysBuildString = new StringBuilder();
        for (int i = 1; i <= keys.size(); i++) {
            keysBuildString.append(String.format(",KEYS[%s]", i));
        }
        final String script = String.format("local result=redis.call('mget' %s) return result",
                keysBuildString.toString());
        byte[][] keyBytes = new byte[keys.size()][];
        for (int i = 0; i < keys.size(); i++) {
            keyBytes[i] = keys.get(i).getBytes();
        }
        List<byte[]> result = this.template.execute(
                (RedisConnection redisConnection) -> redisConnection.eval(script.getBytes(), ReturnType.MULTI, keys.size(), keyBytes));
        return result.stream().filter(Objects::nonNull).map(bytes -> new String(bytes)).collect(Collectors.toList());
    }

    public Long addAllIntoSet(String key, Collection<String> collection) {
        return setOps.add(key, collection.toArray(new String[0]));
    }

    public Boolean isSetMember(String key, String value) {
        return setOps.isMember(key, value);
    }

    public Collection<String> getSetMembers(String key) {
        return setOps.members(key);
    }

    public Set<String> zrangeAll(String key) {
        return this.zrange(key, 0, -1);
    }

    public Set<String> zrange(String key, int start, int end) {
        return zsetOps.range(key, start, end);
    }

    @PostConstruct
    private void init() {
        this.valueOps = this.template.opsForValue();
        this.listOps = this.template.opsForList();
        this.hashOps = this.template.opsForHash();
        this.geoOperations = this.template.opsForGeo();
        this.setOps = this.template.opsForSet();
        this.zsetOps = this.template.opsForZSet();
    }
}